package org.example.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerCallback {
  public static void main(String[] args) throws InterruptedException {
    // 0 配置
    Properties properties = new Properties();
    // 连接集群
    properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "server11:9092");
    // 指定对应的key和value的序列化类型
    properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
    properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());


    // 1.创建kafka生产者对象
    try (KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties)) {


      // 2.发送数据
      for (int i = 0; i < 500; i++) {
        kafkaProducer.send(new ProducerRecord<>("first", "atguigu" + i),
            (metadata, exception) -> {
              if (exception == null) {
                System.out.println("主题：" + metadata.topic() + ", 分区：" + metadata.partition());
              }

            });

        Thread.sleep(1);
      }

      // 3.关闭资源
    }
  }
}
